package com.frank.demo.zookeeper.curator.zk;

import com.alibaba.fastjson.JSONObject;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ManagementFactory;
import java.util.concurrent.TimeUnit;

/**
 * 分布式锁
 * Created by zhy on 2018/4/24.
 */
public final class ZkStarter {
    private static final Logger log = LoggerFactory.getLogger(ZkStarter.class);
    //retryTimes
    private static final int RETRY_TIME = Integer.MAX_VALUE;
    //retryTime
    private static final int RETRY_INTERVAL = 1000;
    //sessionTimeOut
    private static final int SESSION_TIMEOUT = 10000;
    //connectTimeout
    private static final int CONNECT_TIMEOUT = 10000;

    private CuratorFramework client = null;
    private ZkHandle zkHandle;
    private String path = null;

    public ZkStarter(ZkHandle zkHandler) throws Exception {
        this.path = zkHandler.path();
        this.zkHandle = zkHandler;
    }

    public void startup() throws Exception {
        //初始化客户端
        initClient();
        //监听节点下孩子节点变化
        addChildreanListener();
        //创建临时节点
        onSuccess(client);
    }

    private void onSuccess(CuratorFramework client) throws Exception {

        TimeUnit.MILLISECONDS.sleep(1000L);
        String data = getData(client, zkHandle.selfPath());
        JSONObject node = JSONObject.parseObject(data);
        node.put("STATUS", 1);
        setData(client, zkHandle.selfPath(), node);
        addLockNode();

        System.out.println("当前节点 self:" + zkHandle.self());
    }

    /**
     * 监听自己节点
     */
    private void addLockNode() throws Exception {
        NodeCache nodeCache = new NodeCache(client, zkHandle.selfPath(), false);
        NodeCacheListener listener = zkHandle.getNodeChange(nodeCache);
        if (listener != null) {
            nodeCache.getListenable().addListener(listener);
            nodeCache.start();
        }
    }

    private void addChildreanListener() throws Exception {
        PathChildrenCache childrenCache = new PathChildrenCache(client, path, false);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client1, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                    case CHILD_REMOVED:
                        System.out.println("节点变更 : " + data.getPath() + ", type : " + event.getType());
                        zkHandle.routeChange(client1, event);
                        break;
                    case CHILD_UPDATED:
                        System.out.println("节点数据变更 : " + data.getPath() + ", type : " + event.getType());
                        //判断是否自己，如果是自己，开始启动
                        zkHandle.routeDataUpdate(client1, event);
                        break;
                    case INITIALIZED:
                        break;
                    case CONNECTION_LOST:
                    case CONNECTION_RECONNECTED:
                    case CONNECTION_SUSPENDED:
                        System.out.println("特殊事件:" + event.getType());
                        System.exit(-1);
                        break;
                    default:
                        break;
                }
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        log.info("Register zk watcher successfully!");
        childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
    }

    private void initClient() throws Exception {
        if (client == null || client.getState() == CuratorFrameworkState.STOPPED) {
            client = CuratorFrameworkFactory
                    .builder()
                    .connectString(zkHandle.zkAddress())
                    .connectionTimeoutMs(CONNECT_TIMEOUT)
                    .sessionTimeoutMs(SESSION_TIMEOUT)
                    .retryPolicy(new RetryNTimes(RETRY_TIME, RETRY_INTERVAL))//重连
                    .build();
            client.start();

            //创建临时节点
            JSONObject initData = new JSONObject();
            initData.put("NODE", ManagementFactory.getRuntimeMXBean().getName());
            initData.put("STATUS", 0);//初始化开始
            String name = client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path + "/node", initData.toString().getBytes());
            zkHandle.setSelf(name);
        } else {
            log.info("client 已启动");
        }

    }

    public static String getData(CuratorFramework client, String path) throws Exception {
        Stat stat = client.checkExists().forPath(path);
        if (stat == null) {
            return null;
        }
        try {
            byte[] data = client.getData().forPath(path);
            return new String(data, "UTF-8");
        } catch (Exception e) {
            return null;
        }
    }

    public static boolean setData(CuratorFramework client, String path, JSONObject data) throws Exception {
        Stat stat = client.checkExists().forPath(path);
        if (stat == null) {
            return false;
        }
        client.setData().forPath(path, data.toString().getBytes());
        return true;
    }
}
